package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.TableProcess;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.utils.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * @ClassName TableProcessFunction
 * @Description TODO
 * @Author$ 邢家俊
 * @Date 2023-5-4 23:52
 * @Version 1.0
 **/
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private Map<String, TableProcess> configMap = new HashMap<>();

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //预加载配置表中的配置信息，避免主流数据先到，广播流(配置表)数据后到，主流找不到对应的配置信息的情况
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop101:3306/edu_config?" +
                "user=root&password=000000&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        String sql = "select * from edu_config.table_process where sink_type='dim'";
        //创建数据库操作对象
        PreparedStatement ps = conn.prepareStatement(sql);
        //执行SQL语句
        ResultSet rs = ps.executeQuery();
        //获取元数据信息
        ResultSetMetaData metaData = rs.getMetaData();
        //处理结果集
        while(rs.next()){
            //定义一个json对象，用于接收查询的一条结果
            JSONObject jsonObj = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = rs.getObject(i);
                jsonObj.put(columnName,columnValue);
            }
            //将json对象转换为对应的实体类对象
            TableProcess tableProcess = jsonObj.toJavaObject(TableProcess.class);
            //将这条对象放到configMap集合中
            configMap.put(tableProcess.getSourceTable(),tableProcess);
        }

        //释放资源
        rs.close();
        ps.close();
        conn.close();
    }

    //处理主流业务数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // {"database":"edu","xid":3692,"data":{"tm_name":"Redmi1","create_time":"2021-12-14 00:00:00","id":1},"old":{"tm_name":"Redmi"},"commit":true,"type":"update","table":"base_trademark","ts":1681439698}
        // System.out.println(jsonObj);
        //获取操作的业务数据库中的表名
        String tableName = jsonObj.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        //根据处理的表名 到广播状态中获取对应的配置信息
        TableProcess tableProcess = null;

        if((tableProcess = broadcastState.get(tableName))!=null
                ||(tableProcess = configMap.get(tableName))!=null){
            System.out.println(jsonObj);
            //如果配置对象不为空 说明是维度数据  {"tm_name":"aa","create_time":"2021-12-14 00:00:00","logo_url":"aaabbbbccc","id":14}
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");

            //在向下游传递数据前，将不需要传递的属性过滤掉  {"tm_name":"Redmi8","id":1}
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj,sinkColumns);

            //在向下游传递数据前  补充输出的目的地表 {"tm_name":"Redmi5","sink_table":"dim_base_trademark","id":1}
            String sinkTable = tableProcess.getSinkTable();
            dataJsonObj.put("sink_table",sinkTable);

            //补充操作类型
            String type = jsonObj.getString("type");
            dataJsonObj.put("type",type);

            //将维度数据向下游传递
            out.collect(dataJsonObj);
        }
    }

    //过滤掉不需要传递到下游的属性
    //dataJsonObj:  {"tm_name":"aa","create_time":"2021-12-14 00:00:00","logo_url":"aaabbbbccc","id":14}
    //sinkColumns:  id,tm_name
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        //使用逗号分隔出表中的字段
        String[] fieldArr = sinkColumns.split(",");
        List<String> fieldList = Arrays.asList(fieldArr);

        //获取json对象中的所有数据 并对其进行遍历
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry->!fieldList.contains(entry.getKey()));
    }

    //处理广播流配置数据
    //"op":"r": {"before":null,"after":{"source_table":"activity_rule","source_type":"ALL","sink_table":"dim_activity_rule","sink_type":"dim","sink_columns":"id,activity_id,activity_type,condition_amount,condition_num,benefit_amount,benefit_discount,benefit_level","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1681285918409,"snapshot":"false","db":"edu_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1681285918409,"transaction":null}
    //"op":"c"：{"before":null,"after":{"source_table":"aa","source_type":"ALL","sink_table":"dim_aa","sink_type":"dim","sink_columns":"a,bc,c","sink_pk":"a","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1681286000000,"snapshot":"false","db":"edu_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000003","pos":11302896,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1681285999620,"transaction":null}
    //"op":"u"：{"before":{"source_table":"aa","source_type":"ALL","sink_table":"dim_aa","sink_type":"dim","sink_columns":"a,bc,c","sink_pk":"a","sink_extend":null},"after":{"source_table":"aa","source_type":"ALL","sink_table":"dim_aa","sink_type":"dim","sink_columns":"a,bc,c,d","sink_pk":"a","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1681286086000,"snapshot":"false","db":"edu_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000003","pos":11303262,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1681286085644,"transaction":null}
    //"op":"d"：{"before":{"source_table":"aa","source_type":"ALL","sink_table":"dim_aa","sink_type":"dim","sink_columns":"a,bc,c,d","sink_pk":"a","sink_extend":null},"after":null,"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1681286126000,"snapshot":"false","db":"edu_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000003","pos":11303656,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1681286125806,"transaction":null}
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //为了操作方便，将jsonStr转换为json对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //获取操作类型
        String op = jsonObj.getString("op");
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if ("d".equals(op)) {
            // 如果是删除操作，从广播状态中删除掉对应的配置信息
            TableProcess before = jsonObj.getObject("before", TableProcess.class);
            //判断是否删除的是维度配置
            String sinkType = before.getSinkType();
            if ("dim".equals(sinkType)) {
                String sourceTable = before.getSourceTable();
                //根据维度表表名  从广播状态中将对应的配置删除掉
                broadcastState.remove(sourceTable);
                configMap.remove(sourceTable);
            }
        } else {
            // 如果是删除外的其它操作，将读取的配置放到广播状态中
            TableProcess after = jsonObj.getObject("after", TableProcess.class);
            String sinkType = after.getSinkType();
            if ("dim".equals(sinkType)) {
                //在向广播状态中放入数据前，提前创建维度表
                //获取Phoenix中维度表表名
                String sinkTable = after.getSinkTable();
                //获取表中字段
                String sinkColumns = after.getSinkColumns();
                //获取主键
                String pk = after.getSinkPk();
                //获取建表扩展
                String ext = after.getSinkExtend();
                checkTable(sinkTable, sinkColumns, pk, ext);

                //将对应的维度数据放到广播状态中
                String sourceTable = after.getSourceTable();
                broadcastState.put(sourceTable, after);
                configMap.put(sourceTable, after);
            }
        }
    }

    //提前创建维度表
    private void checkTable(String sinkTable, String sinkColumns, String pk, String ext) {
        if (pk == null) {
            pk = "id";
        }
        if (ext == null) {
            ext = "";
        }
        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists "+ EduConfig.PHOENIX_SCHEMA +"." + sinkTable + "(");

        String[] columnArr = sinkColumns.split(",");
        for (int i = 0; i < columnArr.length; i++) {
            String columnName = columnArr[i];
            //判断是否为主键字段
            if(pk.equals(columnName)){
                createSql.append(columnName + " varchar primary key");
            }else{
                createSql.append(columnName + " varchar ");
            }
            if(i < columnArr.length - 1){
                createSql.append(",");
            }
        }
        createSql.append(") " + ext);
        System.out.println("在Phoenix中执行的建表语句:" + createSql);

        //执行建表语句
        PhoenixUtil.executesql(createSql.toString());

    }
}
